/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.store;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.common.constant.LoggerName;

/**
 * 消费队列实现
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-21
 */
public class ConsumeQueue {
	// 存储单元大小
	public static final int CQStoreUnitSize = 20;
	private static final Logger log = LoggerFactory.getLogger(LoggerName.StoreLoggerName);
	private static final Logger logError = LoggerFactory.getLogger(LoggerName.StoreErrorLoggerName);
	// 存储顶层对象
	private final DefaultMessageStore defaultMessageStore;
	// 存储消息索引的队列
	private final MapedFileQueue mapedFileQueue;
	// Topic
	private final String topic;
	// queueId
	private final int queueId;
	// 写索引时用到的ByteBuffer
	private final ByteBuffer byteBufferIndex;
	// 配置
	private final String storePath;
	private final int mapedFileSize;
	// 最后一个消息对应的物理Offset
	private long maxPhysicOffset = -1;
	// 逻辑队列的最小Offset，删除物理文件时，计算出来的最小Offset
	// 实际使用需要除以 StoreUnitSize
	private volatile long minLogicOffset = 0;

	public ConsumeQueue(//
			final String topic, //
			final int queueId, //
			final String storePath, //
			final int mapedFileSize, //
			final DefaultMessageStore defaultMessageStore) {
		this.storePath = storePath;
		this.mapedFileSize = mapedFileSize;
		this.defaultMessageStore = defaultMessageStore;

		this.topic = topic;
		this.queueId = queueId;

		String queueDir = this.storePath//
				+ File.separator + topic//
				+ File.separator + queueId;//

		this.mapedFileQueue = new MapedFileQueue(queueDir, mapedFileSize, null);

		this.byteBufferIndex = ByteBuffer.allocate(CQStoreUnitSize);
	}

	public boolean load() {
		boolean result = this.mapedFileQueue.load();
		log.info("load consume queue " + this.topic + "-" + this.queueId + " " + (result ? "OK" : "Failed"));
		return result;
	}

	/**
	 *============================== 恢复ConsumeQueue内存数据================================================================= 
	 *============================== 1、更新ConsumeQueue.maxPhysicOffset变量（最后一个消息对应的物理offset）====================
	 *===============================2、删除最后一个消息所在文件之后的文件以及对应的MapedFile对象。==============================
	 *在Broker启动过程中会调用该方法
	 */
	public void recover() {
		final List<MapedFile> mapedFiles = this.mapedFileQueue.getMapedFiles();
		if (!mapedFiles.isEmpty()) {
			// 从倒数第三个文件开始恢复
			int index = mapedFiles.size() - 3;
			// 若文件总数没有达到3个，则从第一个文件开始恢复
			if (index < 0)
				index = 0;

			int mapedFileSizeLogics = this.mapedFileSize;
			MapedFile mapedFile = mapedFiles.get(index);
			ByteBuffer byteBuffer = mapedFile.sliceByteBuffer();
			long processOffset = mapedFile.getFileFromOffset();
			long mapedFileOffset = 0;
			while (true) {
				for (int i = 0; i < mapedFileSizeLogics; i += CQStoreUnitSize) {
					long offset = byteBuffer.getLong();
					int size = byteBuffer.getInt();
					long tagsCode = byteBuffer.getLong();

					// 说明当前存储单元有效
					// TODO 这样判断有效是否合理？
					if (offset >= 0 && size > 0) {
						mapedFileOffset = i + CQStoreUnitSize;
						this.maxPhysicOffset = offset;
					} else {
						log.info("recover current consume queue file over,  " + mapedFile.getFileName() + " " + offset + " " + size + " " + tagsCode);
						break;
					}
				}

				// 走到文件末尾，切换至下一个文件
				if (mapedFileOffset == mapedFileSizeLogics) {
					index++;
					if (index >= mapedFiles.size()) {
						// 当前条件分支不可能发生
						log.info("recover last consume queue file over, last maped file " + mapedFile.getFileName());
						break;
					} else {
						mapedFile = mapedFiles.get(index);
						byteBuffer = mapedFile.sliceByteBuffer();
						// 文件起始偏移量
						processOffset = mapedFile.getFileFromOffset();
						mapedFileOffset = 0;
						log.info("recover next consume queue file, " + mapedFile.getFileName());
					}
				} else {
					log.info("recover current consume queue queue over " + mapedFile.getFileName() + " " + (processOffset + mapedFileOffset));
					break;
				}
			}
			// processOffset为该文件起始偏移量
			// mappedFileOffest为最后一个有效的消息单元在该MappedFile对象中的位置偏移量
			processOffset += mapedFileOffset;
			this.mapedFileQueue.truncateDirtyFiles(processOffset);
		}
	}

	/**
	 * 二分查找查找消息发送时间最接近timestamp逻辑队列的offset
	 */
	public long getOffsetInQueueByTime(final long timestamp) {
		// 获取文件的更新时间在该时间戳之后的文件对应的MapedFile对象。
		MapedFile mapedFile = this.mapedFileQueue.getMapedFileByTime(timestamp);
		if (mapedFile != null) {
			long offset = 0;
			// low:第一个索引信息的起始位置
			// minLogicOffset有设置值则从
			// minLogicOffset-mapedFile.getFileFromOffset()位置开始才是有效值
			int low = minLogicOffset > mapedFile.getFileFromOffset() ? (int) (minLogicOffset - mapedFile.getFileFromOffset()) : 0;

			// high:最后一个索引信息的起始位置
			int high = 0;
			int midOffset = -1, targetOffset = -1, leftOffset = -1, rightOffset = -1;
			long leftIndexValue = -1L, rightIndexValue = -1L;

			// 取出该mapedFile里面所有的映射空间(没有映射的空间并不会返回,不会返回文件空洞)
			SelectMapedBufferResult sbr = mapedFile.selectMapedBuffer(0);
			if (null != sbr) {
				ByteBuffer byteBuffer = sbr.getByteBuffer();
				high = byteBuffer.limit() - CQStoreUnitSize;
				try {
					while (high >= low) {
						midOffset = (low + high) / (2 * CQStoreUnitSize) * CQStoreUnitSize;
						byteBuffer.position(midOffset);
						long phyOffset = byteBuffer.getLong();
						int size = byteBuffer.getInt();

						// 比较时间, 折半
						long storeTime = this.defaultMessageStore.getCommitLog().pickupStoretimestamp(phyOffset, size);
						if (storeTime < 0) {
							// 没有从物理文件找到消息，此时直接返回0
							return 0;
						} else if (storeTime == timestamp) {
							targetOffset = midOffset;
							break;
						} else if (storeTime > timestamp) {
							high = midOffset - CQStoreUnitSize;
							rightOffset = midOffset;
							rightIndexValue = storeTime;
						} else {
							low = midOffset + CQStoreUnitSize;
							leftOffset = midOffset;
							leftIndexValue = storeTime;
						}
					}

					if (targetOffset != -1) {
						// 查询的时间正好是消息索引记录写入的时间
						offset = targetOffset;
					} else {
						if (leftIndexValue == -1) {
							// timestamp 时间小于该MapedFile中第一条记录记录的时间
							offset = rightOffset;
						} else if (rightIndexValue == -1) {
							// timestamp 时间大于该MapedFile中最后一条记录记录的时间
							offset = leftOffset;
						} else {
							// 取最接近timestamp的offset
							offset = Math.abs(timestamp - leftIndexValue) > Math.abs(timestamp - rightIndexValue) ? rightOffset : leftOffset;
						}
					}

					return (mapedFile.getFileFromOffset() + offset) / CQStoreUnitSize;
				} finally {
					sbr.release();
				}
			}
		}

		// 映射文件被标记为不可用时返回0
		return 0;
	}

	/**
	 * ============================== 1、删除指定偏移量processOffset所在文件之后的逻辑文件===============================
	 * ============================== 2、修正所在文件对应的MappedFile对象的写入位置和提交位置两个变量。===================
	 * 根据物理Offset删除无效逻辑文件
	 */
	public void truncateDirtyLogicFiles(long phyOffet) {
		// 逻辑队列每个文件大小
		int logicFileSize = this.mapedFileSize;

		// 先改变逻辑队列存储的物理Offset
		this.maxPhysicOffset = phyOffet - 1;

		while (true) {
			MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile2();
			if (mapedFile != null) {
				ByteBuffer byteBuffer = mapedFile.sliceByteBuffer();
				// 先将Offset清空
				mapedFile.setWrotePostion(0);
				mapedFile.setCommittedPosition(0);

				for (int i = 0; i < logicFileSize; i += CQStoreUnitSize) {
					long offset = byteBuffer.getLong();
					int size = byteBuffer.getInt();
					byteBuffer.getLong();

					// 逻辑文件起始单元
					if (0 == i) {
						// 第一块消息单元中的物理偏移值（第1至8个字节）大于了processOffset
						if (offset >= phyOffet) {
							// 则从MapedFile列表中删除该记录并从磁盘中删除物理文件，然后继续获取列表的最后一个文件（即为当初的倒数第二个文件了）并从第一个数据块开始解析
							this.mapedFileQueue.deleteLastMapedFile();
							break;
						} else {
							// 否则不断地更新MappedFile文件wrotePostion和CommittedPosition两个变量为解析到的数据块位置
							int pos = i + CQStoreUnitSize;
							mapedFile.setWrotePostion(pos);
							mapedFile.setCommittedPosition(pos);
							this.maxPhysicOffset = offset;
						}
					}
					// 逻辑文件中间单元
					else {
						// 说明当前存储单元有效
						if (offset >= 0 && size > 0) {
							// 如果逻辑队列存储的最大物理offset大于物理队列最大offset，则返回
							if (offset >= phyOffet) {
								return;
							}

							int pos = i + CQStoreUnitSize;
							mapedFile.setWrotePostion(pos);
							mapedFile.setCommittedPosition(pos);
							this.maxPhysicOffset = offset;

							// 如果最后一个MapedFile扫描完，则返回
							if (pos == logicFileSize) {
								return;
							}
						} else {
							return;
						}
					}
				}
			} else {
				break;
			}
		}
	}

	/**
	 * 返回最后一条消息对应物理队列的下一个偏移量Next Offset
	 */
	public long getLastOffset() {
		// 物理队列Offset
		long lastOffset = -1;
		// 逻辑队列每个文件大小
		int logicFileSize = this.mapedFileSize;
		// 获取列表中的最后一个MapedFile对象（即对应的是最后一个物理文件）
		MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile2();
		if (mapedFile != null) {
			// 找到写入位置对应的索引项的起始位置
			int position = mapedFile.getWrotePostion() - CQStoreUnitSize;
			if (position < 0)
				position = 0;

			ByteBuffer byteBuffer = mapedFile.sliceByteBuffer();
			byteBuffer.position(position);
			for (int i = 0; i < logicFileSize; i += CQStoreUnitSize) {
				long offset = byteBuffer.getLong();
				int size = byteBuffer.getInt();
				byteBuffer.getLong();

				// 说明当前存储单元有效
				if (offset >= 0 && size > 0) {
					lastOffset = offset + size;
				} else {
					break;
				}
			}
		}

		return lastOffset;
	}

	/**
	 * 消息刷盘
	 * ===========================将内存中的消息写入磁盘文件中===================
	 */
	public boolean commit(final int flushLeastPages) {
		return this.mapedFileQueue.commit(flushLeastPages);
	}

	public int deleteExpiredFile(long offset) {
		int cnt = this.mapedFileQueue.deleteExpiredFileByOffset(offset, CQStoreUnitSize);
		// 无论是否删除文件，都需要纠正下最小值，因为有可能物理文件删除了，但是逻辑文件一个也删除不了
		this.correctMinOffset(offset);
		return cnt;
	}

	/**
	 * ====================根据物理队列最小offset计算修正逻辑队列最小offset===============
	 * 逻辑队列的最小Offset要比传入的物理最小phyMinOffset大
	 */
	public void correctMinOffset(long phyMinOffset) {
		// 获取MapedFileQueue队列中第一个MapedFile对象
		MapedFile mapedFile = this.mapedFileQueue.getFirstMapedFileOnLock();
		if (mapedFile != null) {
			// 从第0位开始读取所有数据；
			SelectMapedBufferResult result = mapedFile.selectMapedBuffer(0);
			if (result != null) {
				try {
					// 有消息存在
					for (int i = 0; i < result.getSize(); i += ConsumeQueue.CQStoreUnitSize) {
						long offsetPy = result.getByteBuffer().getLong();
						result.getByteBuffer().getInt();
						result.getByteBuffer().getLong();
						// 检查每个消息单元中保存的物理偏移量（前8字节）大于等于最小物理偏移量
						if (offsetPy >= phyMinOffset) {
							// 将此消息单元的位置加上该文件的起始偏移量作为ConsumeQueue数据的最小逻辑偏移量存于变量minLogicOffset中。
							this.minLogicOffset = result.getMapedFile().getFileFromOffset() + i;
							log.info("compute logics min offset: " + this.getMinOffsetInQuque() + ", topic: " + this.topic + ", queueId: " + this.queueId);
							break;
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					result.release();
				}
			}
		}
	}

	public long getMinOffsetInQuque() {
		return this.minLogicOffset / CQStoreUnitSize;
	}
	/**
	 * =======================将commitlog物理偏移量/消息大小等信息直接写入consumequeue中
	 * @param offset
	 * @param size
	 * @param tagsCode
	 * @param storeTimestamp
	 * @param logicOffset
	 */
	public void putMessagePostionInfoWrapper(long offset, int size, long tagsCode, long storeTimestamp, long logicOffset) {
		final int MaxRetries = 5;
		boolean canWrite = this.defaultMessageStore.getRunningFlags().isWriteable();
		for (int i = 0; i < MaxRetries && canWrite; i++) {
			boolean result = this.putMessagePostionInfo(offset, size, tagsCode, logicOffset);
			if (result) {
				// 更新StoreCheckpoint.logicsMsgTimestamp变量值为请求对象中的storeTimestamp值；
				this.defaultMessageStore.getStoreCheckpoint().setLogicsMsgTimestamp(storeTimestamp);
				return;
			}
			// 只有一种情况会失败，创建新的MapedFile时报错或者超时
			else {
				// XXX: warn and notify me
				log.warn("[BUG]put commit log postion info to " + topic + ":" + queueId + " " + offset + " failed, retry " + i + " times");

				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					log.warn("", e);
				}
			}
		}

		// XXX: warn and notify me
		log.error("[BUG]consume queue can not write, {} {}", this.topic, this.queueId);
		this.defaultMessageStore.getRunningFlags().makeLogicsQueueError();
	}

	/**
	 * 存储一个20字节的信息，putMessagePostionInfo只有一个线程调用，所以不需要加锁
	 * 
	 * @param offset
	 *            消息对应的CommitLog offset
	 * @param size
	 *            消息在CommitLog存储的大小
	 * @param tagsCode
	 *            tags 计算出来的长整数
	 * @return 是否成功
	 */
	private boolean putMessagePostionInfo(final long offset, final int size, final long tagsCode, final long cqOffset) {
		// 在数据恢复时会走到这个流程
		if (offset <= this.maxPhysicOffset) {
			return true;
		}
		// 根据入参信息构建consumequeue中的数据格式buffer块
		this.byteBufferIndex.flip();
		this.byteBufferIndex.limit(CQStoreUnitSize);
		this.byteBufferIndex.putLong(offset);
		this.byteBufferIndex.putInt(size);
		this.byteBufferIndex.putLong(tagsCode);

		final long expectLogicOffset = cqOffset * CQStoreUnitSize;
		// 找到最后一个MappedFile文件
		MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile(expectLogicOffset);
		if (mapedFile != null) {
			// 纠正MapedFile逻辑队列索引顺序
			if (mapedFile.isFirstCreateInQueue() && cqOffset != 0 && mapedFile.getWrotePostion() == 0) {
				this.minLogicOffset = expectLogicOffset;
				this.fillPreBlank(mapedFile, expectLogicOffset);
				log.info("fill pre blank space " + mapedFile.getFileName() + " " + expectLogicOffset + " " + mapedFile.getWrotePostion());
			}

			if (cqOffset != 0) {
				long currentLogicOffset = mapedFile.getWrotePostion() + mapedFile.getFileFromOffset();
				if (expectLogicOffset != currentLogicOffset) {
					// XXX: warn and notify me
					logError.warn("[BUG]logic queue order maybe wrong, expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}", //
							expectLogicOffset, //
							currentLogicOffset, //
							this.topic, //
							this.queueId, //
							expectLogicOffset - currentLogicOffset//
					);
				}
			}

			// 记录物理队列最大offset
			this.maxPhysicOffset = offset;
			// 将consumequeue的格式数据追加到MappedFile文件的后面
			return mapedFile.appendMessage(this.byteBufferIndex.array());
		}

		return false;
	}

	private void fillPreBlank(final MapedFile mapedFile, final long untilWhere) {
		ByteBuffer byteBuffer = ByteBuffer.allocate(CQStoreUnitSize);
		byteBuffer.putLong(0L);
		byteBuffer.putInt(Integer.MAX_VALUE);
		byteBuffer.putLong(0L);

		int until = (int) (untilWhere % this.mapedFileQueue.getMapedFileSize());
		for (int i = 0; i < until; i += CQStoreUnitSize) {
			mapedFile.appendMessage(byteBuffer.array());
		}
	}

	/**
	 * ==========================根据消息序号索引获取consumequeue数据======================
	 * 返回Index Buffer
	 * 
	 * @param startIndex
	 *            起始偏移量索引
	 */
	public SelectMapedBufferResult getIndexBuffer(final long startIndex) {
		int mapedFileSize = this.mapedFileSize;
		// 消息单元的序号*每条数据的大小=起始偏移量offset
		long offset = startIndex * CQStoreUnitSize;
		if (offset >= this.getMinLogicOffset()) {
			// 根据该起始偏移量找到所在的文件，即对应的MapedFile对象
			MapedFile mapedFile = this.mapedFileQueue.findMapedFileByOffset(offset);
			if (mapedFile != null) {
				SelectMapedBufferResult result = mapedFile.selectMapedBuffer((int) (offset % mapedFileSize));
				return result;
			}
		}
		return null;
	}

	/**
	 * ====================获取指定位置所在文件的下一个文件的起始偏移量===================
	 * @param index
	 * @return
	 */
	public long rollNextFile(final long index) {
		int mapedFileSize = this.mapedFileSize;
		int totalUnitsInFile = mapedFileSize / CQStoreUnitSize;
		return (index + totalUnitsInFile - index % totalUnitsInFile);
	}

	public String getTopic() {
		return topic;
	}

	public int getQueueId() {
		return queueId;
	}

	public long getMaxPhysicOffset() {
		return maxPhysicOffset;
	}

	public void setMaxPhysicOffset(long maxPhysicOffset) {
		this.maxPhysicOffset = maxPhysicOffset;
	}

	public void destroy() {
		this.maxPhysicOffset = -1;
		this.minLogicOffset = 0;
		this.mapedFileQueue.destroy();
	}

	public long getMinLogicOffset() {
		return minLogicOffset;
	}

	public void setMinLogicOffset(long minLogicOffset) {
		this.minLogicOffset = minLogicOffset;
	}

	/**
	 * 获取当前队列中的消息总数
	 */
	public long getMessageTotalInQueue() {
		return this.getMaxOffsetInQuque() - this.getMinOffsetInQuque();
	}

	public long getMaxOffsetInQuque() {
		return this.mapedFileQueue.getMaxOffset() / CQStoreUnitSize;
	}
}
